package com.langerra.client.channel;

import java.io.Serializable;
import java.util.LinkedList;

import com.langerra.shared.channel.InputChannel;
import com.google.gwt.user.client.Timer;
import com.google.gwt.user.client.rpc.AsyncCallback;

public abstract class AsyncChannelScheduler<T extends Serializable> {

	// amount of time used for each message dispatch process. if there are more
	// messages to dispatch however this time has elapsed the loop stops and
	// waits for EPOC to elapse before it restarts again
	// EPOC is the relaxation time between each dispatch process

	private final int quanta;
	private final int epoc;
	private final LinkedList<T> buffer;
	private final Timer processor = new Timer() {

		@Override
		public synchronized void run() {

			final LinkedList<T> messages = buffer;
			final long start = System.currentTimeMillis();

			while (!messages.isEmpty()) {

				T message = messages.poll();
				if (message != null)
					async(message);

				// if we have used more than the quanta time, stop the current
				// processing and schedule another run later on.
				final long now = System.currentTimeMillis();
				if (start + quanta < now) {
					break;
				}
			}

			if (!messages.isEmpty())
				processor.schedule(epoc);
		}
	};

	private void schedule() {
		processor.schedule(epoc);
	}

	private void add(
			T message) {
		buffer.add(message);

		if (buffer.size() == 1)
			schedule();
	}

	protected abstract void async(
			T message);

	public AsyncChannelScheduler(
			InputChannel<T> channel) {
		this(channel, 200, 50);
	}

	public AsyncChannelScheduler(
			InputChannel<T> channel,
			int quanta,
			int epoc) {
		this.quanta = quanta;
		this.epoc = epoc;
		this.buffer = new LinkedList<T>();
		channel.async(new AsyncCallback<T>() {

			@Override
			public void onSuccess(
					T result) {
				add(result);
			}

			@Override
			public void onFailure(
					Throwable caught) {
			}
		});
	}

	public LinkedList<T> getBuffer() {
		return buffer;
	}

}
